package com.neusoft.myaggregation;

import com.neusoft.pojo.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.connector.source.util.ratelimit.RateLimiterStrategy;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.connector.datagen.source.DataGeneratorSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.Random;

public class AggregationApiMain {
    private static final Random random = new Random();


    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
//        定义数据源
        DataGeneratorSource<String> dataGeneratorSource = new DataGeneratorSource(
                l -> {
                    String line = ("" + (random.nextInt(3) + 1) + "," + System.currentTimeMillis() + "," + random.nextInt(90, 100));
                    return line;
                },
                Long.MAX_VALUE,
                RateLimiterStrategy.perSecond(1),
                Types.STRING);
        env.fromSource(dataGeneratorSource, WatermarkStrategy.noWatermarks(), "dataGeneratorSource")
                .map(new MapFunction<String, Tuple3<String, Long, Integer>>() {

                    @Override
                    public Tuple3<String, Long, Integer> map(String value) throws Exception {
                        String[] words = value.split(",");
                        return Tuple3.of(words[0], Long.parseLong(words[1]), Integer.parseInt(words[2]));

                    }
                })
                .keyBy(t -> t.f0)
                .reduce((a, b) -> Tuple3.of(a.f0, a.f1, a.f2 + b.f2))
                .print();
        env.execute();
    }


    public static void main2(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
//        定义数据源
        DataGeneratorSource<String> dataGeneratorSource = new DataGeneratorSource(
                l -> {
                    String line = ("" + (random.nextInt(3) + 1) + "," + System.currentTimeMillis() + "," + random.nextInt(90, 100));
                    return line;
                },
                Long.MAX_VALUE,
                RateLimiterStrategy.perSecond(1),
                Types.STRING);
        env.fromSource(dataGeneratorSource, WatermarkStrategy.noWatermarks(), "dataGeneratorSource")
                .map(new MapFunction<String, Tuple3<String, Long, Integer>>() {

                    @Override
                    public Tuple3<String, Long, Integer> map(String value) throws Exception {
                        String[] words = value.split(",");
                        return Tuple3.of(words[0], Long.parseLong(words[1]), Integer.parseInt(words[2]));

                    }
                })
                .keyBy(t -> t.f0)
                .max(2)
                .print();
        env.execute();
    }


    public static void main1(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
//        定义数据源
        DataGeneratorSource<String> dataGeneratorSource = new DataGeneratorSource(
                l -> {
                    String line = ("" + (random.nextInt(3) + 1) + "," + System.currentTimeMillis() + "," + random.nextInt(90, 100));
                    return line;
                },
                Long.MAX_VALUE,
                RateLimiterStrategy.perSecond(1),
                Types.STRING);
        env.fromSource(dataGeneratorSource, WatermarkStrategy.noWatermarks(), "dataGeneratorSource")
                .map(new MapFunction<String, WaterSensor>() {

                    @Override
                    public WaterSensor map(String value) throws Exception {
                        String[] words = value.split(",");
                        return new WaterSensor(words[0], Long.parseLong(words[1]), Integer.parseInt(words[2]));
                    }
                })
                .keyBy(ws -> ws.getId())
                .max("vc")
                .print();
        env.execute();
    }


}
